package vn.com.epi.re.log.Import;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import vn.com.epi.re.ApplicationContext;
import vn.com.epi.re.data.ArticleRowData;
import vn.com.epi.re.data.UserReading;
import vn.com.epi.re.table.ArticleTable;
import vn.com.epi.re.table.UserTable;
import vn.com.epi.re.tools.StringUtils;
import vn.com.epi.re.tools.TableUtils;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.io.Closeables;

public class LogImporter extends Configured implements Tool {
    private final Log log = LogFactory.getLog(LogImporter.class);
    public static final int GROUP_SIZE = 1000;
    private ArrayListMultimap<String, UserReading> uMap;
    private Map<Integer, ArticleRowData> aMap;
    
    public LogImporter() {
        uMap = ArrayListMultimap.create();
        aMap = Maps.newHashMap();
    }
    
    @SuppressWarnings("unused")
    private boolean validLog(String logLine) {
        String[] chunks = logLine.split(",");
        if (chunks.length != 5)
            return false;
        try {
            long timestamp = Long.parseLong(chunks[0]);
            int preference = Integer.parseInt(chunks[2].substring(3, chunks[2].length()));
            int category = Integer.parseInt(chunks[4]);
        } catch (Exception e) {
            return false;
        }
        return true;
    }
    
    private boolean validUser(String userId) {
        if ((userId.matches("[0-9]+")) && (userId.length() == 15))
            return true;
        if ((userId.length() == 44) && (userId.startsWith("WP8-")))
            return true;
        if ((userId.length() == 36) && (!userId.startsWith("RANDOM")))
            return true;
        return false;
    }
    
    private void loadData(String logFile) throws IOException {
        BufferedReader br = null;
        try {
            br = new BufferedReader(new FileReader(logFile));
            String logLine;
            while ((logLine = br.readLine()) != null) {
                if (validLog(logLine)) {
                    String[] chunks = logLine.split(",");
                    String user = chunks[1];
                    // only using ios and android log
                    if (!validUser(user))
                        continue;
                    int article = Integer.parseInt(chunks[3]);
                    UserReading userRow = new UserReading(logLine);
                    ArticleRowData articleRow = new ArticleRowData(logLine);
                    uMap.put(user, userRow);
                    if (!aMap.containsKey(article)) 
                        aMap.put(article, articleRow);
                    else {
                        int totalCount = aMap.get(article).getTotalCount() + 1;
                        long timestamp = aMap.get(article).getTimestamp();
                        int categoryId = aMap.get(article).getCategoryId();
                        if (articleRow.getCategoryId() == 0 && categoryId > 0)
                            articleRow.setCategoryId(categoryId);
                        articleRow.setTotalCount(totalCount);
                        articleRow.setTimestamp(Math.min(articleRow.getTimestamp(), timestamp));
                        aMap.put(article, articleRow);
                    }
                }
            }                   
        } catch (Exception e) {
            System.out.println("Error while load data from file: " + e.toString());
            System.exit(-1);
        } finally {
            if (br != null)
                try {
                    br.close();
                } catch (Exception ex) {                    
                }
        }
    }
    
    private void importUser() throws IOException {
        HTableInterface table = ApplicationContext.get().getTable(UserTable.NAME);
        List<Put> puts = Lists.newArrayList();
        List<Get> gets = Lists.newArrayList();
        int countDown = uMap.keySet().size();
        int count = 0;
        try {            
            for (String user : uMap.keySet()) {
                byte[] row = Bytes.toBytes(user);
                Get get = new Get(row);
                gets.add(get);
                countDown--;
                count++;                
                if (count == GROUP_SIZE || countDown == 0) {
                    Result[] results = table.get(gets);
                    for (int i = 0; i < results.length; i++) {
                        Map<Integer, Boolean> visited = new HashMap<Integer, Boolean>();
                        String userId = Bytes.toString(results[i].getRow());
                        StringBuilder timestampSb = new StringBuilder();
                        StringBuilder preferenceSb = new StringBuilder();
                        StringBuilder articleSb = new StringBuilder();
                        if (userId != null) {
                            timestampSb.append(UserTable.getVisitedTimestamps(results[i]));
                            preferenceSb.append(UserTable.getReadingPreferences(results[i]));
                            articleSb.append(UserTable.getVisitedArticles(results[i]));
                            if (articleSb.length() > 0) {
                                String[] chunks = articleSb.toString().split(",");
                                for (int j = 0; j < chunks.length; j++) {
                                    visited.put(Integer.parseInt(chunks[j]), true);
                                }
                            }
                        } else
                            userId = Bytes.toString(gets.get(i).getRow());
                        List<UserReading> list = uMap.get(userId);
                        StringUtils.insertCommaChar(timestampSb);
                        StringUtils.insertCommaChar(preferenceSb);
                        StringUtils.insertCommaChar(articleSb);
                        for (UserReading record : list)  
                            if (!visited.containsKey(record.getArticleId())) {
                                visited.put(record.getArticleId(), true);
                                timestampSb.append(record.getTimestamp() + ",");                                
                                preferenceSb.append(record.getPreference() + ",");
                                articleSb.append(record.getArticleId() + ",");
                            }
                        StringUtils.deleteLastChar(timestampSb);
                        StringUtils.deleteLastChar(preferenceSb);
                        StringUtils.deleteLastChar(articleSb);
                        Put put = new Put(Bytes.toBytes(userId));
                        put.add(UserTable.FAM_MAIN, UserTable.COL_VISITED_TIMESTAMPS, Bytes.toBytes(timestampSb.toString()));
                        put.add(UserTable.FAM_MAIN, UserTable.COL_READING_PREFERENCES, Bytes.toBytes(preferenceSb.toString()));
                        put.add(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES, Bytes.toBytes(articleSb.toString()));
                        puts.add(put);                        
                    }
                    table.put(puts);
                    puts = Lists.newArrayList();
                    gets = Lists.newArrayList();
                    count = 0;
                }
            }
        } catch (Exception e) {
            System.out.println("Error while import user: " + e.toString());
            System.exit(-1);
        } finally {
            TableUtils.closeQuietly(table);
            uMap = null;
        }        
    }
    
    private void importArticle() throws IOException {
        HTableInterface table = ApplicationContext.get().getTable(ArticleTable.NAME);
        List<Put> puts = Lists.newArrayList();
        List<Get> gets = Lists.newArrayList();
        int countDown = aMap.size();
        int count = 0;
        try {
            for (int article : aMap.keySet()) {
                byte[] row = Bytes.toBytes(article);
                Get get = new Get(row);
                gets.add(get);
                countDown--;
                count++;
                if (count == GROUP_SIZE || countDown == 0) {
                    Result[] results = table.get(gets);
                    for (int i = 0; i < results.length; i++) {
                        int articleId;
                        int totalCount = 0;
                        long timestamp = Integer.MAX_VALUE;
                        int categoryId;
                        int oldCategoryId = -1;
                        if (results[i].getRow() != null) {
                            articleId = Bytes.toInt(results[i].getRow());
                            timestamp  = ArticleTable.getFirstVisitTimestamp(results[i]);
                            oldCategoryId = ArticleTable.getCategoryId(results[i]);
                            totalCount = ArticleTable.getTotalCount(results[i]);
                        } else
                            articleId = Bytes.toInt(gets.get(i).getRow());
                        ArticleRowData articleRow = aMap.get(articleId);
                        timestamp = Math.min(timestamp, articleRow.getTimestamp());
                        categoryId = articleRow.getCategoryId();
                        totalCount += articleRow.getTotalCount();
                        Put put = new Put(Bytes.toBytes(articleId));
                        put.add(ArticleTable.FAM_MAIN, ArticleTable.COL_FIRST_VISIT_TIMESTAMPS, Bytes.toBytes(timestamp));
                        if (oldCategoryId == -1)
                            put.add(ArticleTable.FAM_MAIN, ArticleTable.COL_CATEGORY_ID, Bytes.toBytes(categoryId));
                        else {
                            if (oldCategoryId == 0)
                                put.add(ArticleTable.FAM_MAIN, ArticleTable.COL_CATEGORY_ID, Bytes.toBytes(categoryId));
                        }
                        put.add(ArticleTable.FAM_MAIN, ArticleTable.COL_TOTAL_COUNT, Bytes.toBytes(totalCount));
                        puts.add(put);
                    }
                    table.put(puts);
                    puts = Lists.newArrayList();
                    gets = Lists.newArrayList();
                    count = 0;
                }                
            }
        } catch (Exception e) {
            System.out.println("Error while import article: " + e.toString());
            System.exit(-1);
        } finally {
            TableUtils.closeQuietly(table);
            aMap = null;
        }        
    }
    
    public void importLog(String logFile) throws IOException {
        log.info("----------------------* start import log *----------------------");
        log.info("step 1: load data to hash map");
        try {
            loadData(logFile);
        } catch (IOException e) {
            log.info("error: log file not found!!!");
            log.info(e.toString());
            return;
        }
        log.info("completed load data!!!");
        log.info("step 2: start import log to hbase");
        log.info("---------* import user *----------");
        importUser();        
        log.info("completed import user");
        log.info("---------* import article *----------");
        importArticle();
        log.info("completed import article");
    }
    
    @Override
    public int run(String[] args) throws Exception {
        importLog(args[0]);
        return 0;
    }       

    public static void main(String[] args) throws Exception {
        long startTime = System.currentTimeMillis();
        ApplicationContext c = ApplicationContext.init();
        try {
            ToolRunner.run(new LogImporter(), args);
        } finally {
            Closeables.closeQuietly(c);            
        }        
        long endTime = System.currentTimeMillis();
        System.out.println("total time used for import log: " + (endTime - startTime) + " ms.");
    }
}
